package com.cui.project.mq;

import com.api.common.alert.CommonAlertHelper;
import com.api.common.common.ErrorCode;
import com.api.common.constant.APIInterfaceMqConstant;
import com.api.common.model.entity.GatewayAccessLog;
import com.api.common.model.entity.InterfaceInfoLog;
import com.cui.project.exception.BusinessException;
import com.cui.project.service.GatewayAccessLogService;
import com.cui.project.service.InterfaceInfoLogService;
import com.google.gson.Gson;
import com.google.gson.JsonSyntaxException;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author: 崔老爷
 * Date: 2024/10/16 11:31
 * @Description:
 */

@Component
@RequiredArgsConstructor
@Slf4j
public class APIMessageConsumer {



    private final InterfaceInfoLogService interfaceInfoLogService;
    private final GatewayAccessLogService gatewayAccessLogService;
    private final CommonAlertHelper alertHelper;

    @RabbitListener(queues = {APIInterfaceMqConstant.API_INTERFACE_LOG_QUEUE}, ackMode = "MANUAL")
    public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        log.info("开始处理接口日志消息: exchange={} queue={} route_key={}, deliveryTag={}, messageLength={}",
                APIInterfaceMqConstant.API_INTERFACE_LOG_EXCHANGE,
                APIInterfaceMqConstant.API_INTERFACE_LOG_QUEUE,
                APIInterfaceMqConstant.API_INTERFACE_LOG_ROUTE_KEY,
                deliveryTag,
                message != null ? message.length() : 0);
        if (StringUtils.isBlank(message)) {
            //空的消息没有必要再放到队列中
            channel.basicNack(deliveryTag, false, false);
            throw new BusinessException(ErrorCode.OPERATION_ERROR, "消息为空");
        }
        try{
            Gson gson = new Gson();
            InterfaceInfoLog interfaceInfoLog = gson.fromJson(message, InterfaceInfoLog.class);
            if(interfaceInfoLog==null){
                channel.basicNack(deliveryTag, false, false);
                throw new BusinessException(ErrorCode.OPERATION_ERROR, "消息为空");
            }
            if(!interfaceInfoLogService.save(interfaceInfoLog)){
                channel.basicNack(deliveryTag, false, false);
                throw new BusinessException(ErrorCode.SYSTEM_ERROR, "保存失败");
            }
            channel.basicAck(deliveryTag,false);
        }catch (Exception e){
            log.error("消费接口日志消息失败: deliveryTag={}, message={}", deliveryTag, message, e);
            // 判断异常类型，决定是否重入队列
            if (e instanceof JsonSyntaxException) {
                alertHelper.dataFormatError(
                        "InterfaceInfoLog",
                        "JSON",
                        "解析失败"
                );
                // 格式错误，不应重试
                channel.basicNack(deliveryTag, false, false);
            } else {
                alertHelper.mqConsumeFailure(
                        APIInterfaceMqConstant.API_INTERFACE_LOG_QUEUE,
                        String.valueOf(deliveryTag),
                        e.getMessage(),
                        "195883938@qq.com"  // 运维团队邮箱
                );
                // 业务异常，可以重试（设置最大重试次数）
                channel.basicNack(deliveryTag, false, true);
            }

            // 或者配置死信队列（DLX）
            // 不要再次抛出异常，否则会导致消息消费线程中断
//            throw new BusinessException(ErrorCode.SYSTEM_ERROR);
        }

    }

    /**
     * 消费api_gateway_access_log_queue的信息
     */
    @RabbitListener(queues = {APIInterfaceMqConstant.API_GATEWAY_ACCESS_LOG_QUEUE},ackMode = "MANUAL")
    public void receiveAALMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        log.info("开始处理网关日志消息: exchange={} queue={} route_key={}, deliveryTag={}, messageLength={}",
                APIInterfaceMqConstant.API_GATEWAY_ACCESS_LOG_EXCHANGE,
                APIInterfaceMqConstant.API_GATEWAY_ACCESS_LOG_QUEUE,
                APIInterfaceMqConstant.API_GATEWAY_ACCESS_LOG_ROUTE_KEY,
                deliveryTag,
                message != null ? message.length() : 0);

        if(StringUtils.isBlank(message)){
            channel.basicNack(deliveryTag,false,false);
            throw new BusinessException(ErrorCode.OPERATION_ERROR,"消息为空");
        }

        try {
            Gson gson=new Gson();
            GatewayAccessLog gatewayAccessLog = gson.fromJson(message, GatewayAccessLog.class);
            if(gatewayAccessLog==null){
                channel.basicNack(deliveryTag,false,false);
                throw new BusinessException(ErrorCode.OPERATION_ERROR,"消息为空");
            }
            if(!gatewayAccessLogService.save(gatewayAccessLog)){
                channel.basicNack(deliveryTag,false,false);
                throw new BusinessException(ErrorCode.SYSTEM_ERROR,"保存失败");
            }
            channel.basicAck(deliveryTag,false);
        }catch (Exception e){
            log.error("消费网关日志消息失败: deliveryTag={}, message={}", deliveryTag, message, e);
            // 判断异常类型，决定是否重入队列
            if (e instanceof JsonSyntaxException) {
                log.warn("数据格式错误告警");
                alertHelper.dataFormatError(
                        "GatewayAccessLog",
                        "JSON",
                        "解析失败"
                );
                // 格式错误，不应重试
                channel.basicNack(deliveryTag, false, false);
            } else {
                log.warn("MQ消息消费失败告警（指定邮箱）");
                alertHelper.mqConsumeFailure(
                        APIInterfaceMqConstant.API_GATEWAY_ACCESS_LOG_QUEUE,
                        String.valueOf(deliveryTag),
                        e.getMessage(),
                        "195883938@qq.com"  // 运维团队邮箱
                );
                // 业务异常，可以重试（设置最大重试次数）
                channel.basicNack(deliveryTag, false, false);
            }

            // 或者配置死信队列（DLX）
            // 不要再次抛出异常，否则会导致消息消费线程中断
//            throw new BusinessException(ErrorCode.SYSTEM_ERROR);
        }

    }


}
